#include "lp_global_vars.h"
#include	"config_parser.h"

int last = 0;
int acceptor_asked_count = 0;

//------------------------------------------------------------------------------
double getMiliseconds() {
	//--------------------
	double msec;
	struct timeval mtv;
	//--------------------
	gettimeofday(&mtv, NULL);
	msec = 1000 * mtv.tv_sec + mtv.tv_usec / (double) 1000;
	//printf("TIME: %f\n", msec);
	return msec;
}

void learner_deliver() {
	int i, j;
	bool flag;
	int status = 0;
	int gap_length = 0;
	gap g;
	for (i = lowestDelivered + 1; i < lp_interval_performance->highest_deliverable_inst_id + 1; i++) {
		if (data[(i % INTERNAL_BUFFER_SIZE)].received) {
			if (!data[(i % INTERNAL_BUFFER_SIZE)].deliver) {
				if (data[(i % INTERNAL_BUFFER_SIZE)].packet.pack_id == (i + 1)) {

					lp_interval_performance->interval_latency += getMiliseconds() - data[(i % INTERNAL_BUFFER_SIZE)].packet.p_msgs[0].mili_seconds;
					flag = true;

					data[(i % INTERNAL_BUFFER_SIZE)].dec = data[(i % INTERNAL_BUFFER_SIZE)].packet.pack_id;
					data[(i % INTERNAL_BUFFER_SIZE)].deliver = true;
					data[(i % INTERNAL_BUFFER_SIZE)].received = false;
					data[(i % INTERNAL_BUFFER_SIZE)].deliver = false;

					data[(i % INTERNAL_BUFFER_SIZE)].packet.p_msgs[0].msg_id = 0;
					data[(i % INTERNAL_BUFFER_SIZE)].packet.pack_id = 0;
					lp_interval_performance->total_num_packets_delivered++;
					lp_interval_performance->interval_num_packets_delivered++;
				}
			}
		} else {
			g.start = i;
			g.end = i;
			gap_length = 0;
			if (g.start > last) {
				for (j = i + 1; j < lp_interval_performance->highest_deliverable_inst_id; j++) {
					if (gap_length < MAX_GAP_LENGTH) {
						if (data[(j % INTERNAL_BUFFER_SIZE)].received)
							break;
						else {
							g.end = j;
							gap_length++;
						}
					}
				}
				if (last != g.end) {
					last = g.end;
					lp_interval_performance->interval_num_packets_lost++;
					//	printf("SENDING: PROCESS: %d,  HD:%d, %d, gap:start:%d, gapEnd:%d\n", learnerID, lp_interval_performance->highest_deliverable_inst_id, lp_interval_performance->total_num_packets_delivered, g.start, g.end);
					status = send(learner_to_acceptor_sid, &g, sizeof(gap), MSG_WAITALL);
					assert(status == sizeof(gap));
				}
			}
			break;
		}
	}
	lowestDelivered = i - 1;
}
//------------------------------------------------------------------------------
void recv_new_packet_from_predecessor(int sd, short event, void * arg) {

	UNUSED_ARG(event);
	UNUSED_ARG(arg);
	//printf("%d\n", sd);
	int status = -1;
	unsigned int len = sizeof(struct sockaddr);

	status = recvfrom(sd, new_data, sizeof(inst), MSG_WAITALL, (struct sockaddr *) &(get_config_info()->learner_infos[get_config_info()->self_learner_id].ring_self_address), &len);
	assert(status == sizeof(inst));

	r_tag = true;
	//new_data = &packet_batch.batch[i];
	if (lp_interval_performance->highest_deliverable_inst_id != new_data->highestDeliverable) {
		lp_interval_performance->highest_deliverable_inst_id = new_data->highestDeliverable;
		learner_deliver();
	}
	if (data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].packet.pack_id == new_data->packet.pack_id) {
		lp_interval_performance->interval_num_redundant_packets_received++;
		lp_interval_performance->total_num_redundant_packets_received++;
	} else {
		lp_interval_performance->interval_num_packets_received++;
		lp_interval_performance->total_num_packets_received++;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].received = true;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].deliver = false;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].instance_id = new_data->instance_id;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].packet.pack_id = new_data->packet.pack_id;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].coordinator_id = new_data->coordinator_id;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].dec = new_data->dec;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].highestDeliverable = -1;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].packet.p_msgs[0] = new_data->packet.p_msgs[0];

	}
	//forward:
	if (get_config_info()->self_learner_id != get_config_info()->total_num_learners - 1) {
		struct sockaddr_in successor_learner__addr;
		unsigned int slen = sizeof(successor_learner__addr);

		status = sendto(get_config_info()->learner_infos[get_config_info()->self_learner_id].successor_sid, new_data, sizeof(inst), 0, (struct sockaddr*) &(get_config_info()->learner_infos[get_config_info()->self_learner_id].ring_successor_address), slen);
		assert(status == sizeof(inst));
	}

}
//------------------------------------------------------------------------------
void recv_lost_packet(int sd, short event, void *arg) {
	int status;
	inst temp;
	memset(&temp, '\0', sizeof(inst));
	status = recv(sd, &temp, sizeof(inst), MSG_WAITALL);
	assert(status > 0);
	if (status == sizeof(inst)) {
		if ((temp.instance_id == (temp.packet.pack_id - 1))) {
			acceptor_asked_count++;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].received = true;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].deliver = false;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].instance_id = temp.instance_id;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].packet.pack_id = temp.packet.pack_id;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].coordinator_id = temp.coordinator_id;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].dec = temp.dec;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].highestDeliverable = -1;
			data[(temp.instance_id % INTERNAL_BUFFER_SIZE)].packet.p_msgs[0] = temp.packet.p_msgs[0];
			printf("RECEIVING: %d\n", temp.packet.pack_id);
		} else {
			acceptor_asked_count++;
			printf("%d is OVERWRITTEN already\n", temp.instance_id);
			lp_interval_performance->interval_num_overwritten_insts++;
			lp_interval_performance->total_num_overwritten_insts++;
			data[((temp.instance_id) % INTERNAL_BUFFER_SIZE)].received = true;
			data[((temp.instance_id) % INTERNAL_BUFFER_SIZE)].deliver = true;
		}
	}
}

//------------------------------------------------------------------------------

